package com.example.redisdemo.configuration;


import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;


/**
 * @author hexinglin
 * @created 2021-08-11 14:16
 **/
/**
 * 定义队列名和交换机
 */
@Configuration
@Slf4j
public class RabbitConfig {

    /**
     * 交换机名称
     */
    public static final String DIRECT_EXCHANGE_NAME = "direct_exchange";
    public static final String DELAY_EXCHANGE_NAME = "delay_exchange";
    public static final String RECEIVE_EXCHANGE = "receive_exchange";

    /**
     * 绑定key，交换机绑定队列时需要指定
     */
    public static final String BINGDING_KEY_TEST1 = "direct_key1";
    public static final String BINGDING_KEY_TEST2 = "direct_key2";
    public static final String DELAY_KEY = "delay_key";

    /**
     * 队列名称
     */
    public static final String QUEUE_TEST1 = "test1";
    public static final String QUEUE_TEST2 = "test2";
    public static final String DELAY_QUEUE = "delay_queue";
    public static final String RECEIVE_QUEUE = "receive_queue";

    /**
     * 构建DirectExchange交换机
     *
     * @return
     */
    @Bean
    public DirectExchange directExchange() {
        // 支持持久化，长期不用补删除
        return new DirectExchange(DIRECT_EXCHANGE_NAME, true, false);
    }



    /**
     * 构建序列
     *
     * @return
     */
    @Bean
    public Queue test1Queue() {
        // 支持持久化
        return new Queue(QUEUE_TEST1, true);
    }

    @Bean
    public Queue test2Queue() {
        // 支持持久化
        return new Queue(QUEUE_TEST2, true);
    }
    /**
     * 绑定交交换机和
     *
     * @return
     */
    @Bean
    public Binding test1Binding() {
        return BindingBuilder.bind(test1Queue()).to(directExchange()).with(BINGDING_KEY_TEST1);
    }

    @Bean
    public Binding test2Binding() {
        return BindingBuilder.bind(test2Queue()).to(directExchange()).with(BINGDING_KEY_TEST2);
    }
    /**
     * 死信队列
     * @return
     */
    @Bean
    public Queue delayQueue(){
        Map<String,Object> map = new HashMap<>(16);
        map.put("x-dead-letter-exchange","receive_exchange");
        map.put("x-dead-letter-routing-key", "receive_key");
        map.put("x-message-ttl", 5*1000);
        return new Queue(DELAY_QUEUE,true,false,false,map);
    }
    @Bean
    public DirectExchange delayExchange() {
        return new DirectExchange(DELAY_EXCHANGE_NAME);
    }
    /**
     * 给死信队列绑定交换机
     * @return
     */
    @Bean
    public Binding delayBinding(){
        return BindingBuilder.bind(delayQueue()).to(delayExchange()).with(DELAY_KEY);
    }

    /**
     * 死信接收交换机
     * @return
     */
    @Bean
    public DirectExchange receiveExchange(){
        return new DirectExchange(RECEIVE_EXCHANGE);
    }

    /**
     * 死信接收队列
     * @return
     */
    @Bean
    public Queue receiveQueue(){
        return new Queue(RECEIVE_QUEUE);
    }

    /**
     * 死信交换机绑定消费队列
     * @return
     */
    @Bean
    public Binding receiveBinding(){
        return BindingBuilder.bind(receiveQueue()).to(receiveExchange()).with("receive_key");
    }



    /**
     * 实例化操作模板
     *
     * @param connectionFactory
     * @return
     */
   /* @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        //必须为true,否则无法触发returnedMessage回调，消息丢失
        rabbitTemplate.setMandatory(true);
        return rabbitTemplate;
    }*/
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(converter());

        // 消息是否成功发送到Exchange
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                String msgId = correlationData.getId();
                log.info("消息成功发送到Exchange,msgid:{}",msgId);

            } else {
                log.info("消息发送到Exchange失败, {}, cause: {}", correlationData, cause);
            }
        });

        // 触发setReturnCallback回调必须设置mandatory=true, 否则Exchange没有找到Queue就会丢弃掉消息, 而不会触发回调
        rabbitTemplate.setMandatory(true);
        // 消息是否从Exchange路由到Queue, 注意: 这是一个失败回调, 只有消息从Exchange路由到Queue失败才会回调这个方法
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            log.info("消息从Exchange路由到Queue失败: exchange: {}, route: {}, replyCode: {}, replyText: {}, message: {}", exchange, routingKey, replyCode, replyText, message);
        });

        return rabbitTemplate;
    }

    @Bean
    public Jackson2JsonMessageConverter converter() {
        return new Jackson2JsonMessageConverter();
    }

}


